# coding: utf-8
'''
Description: 
version: 1.0.0
Author: peng2.zhang
Date: 2021-09-01 14:52:33
LastEditors: peng2.zhang
LastEditTime: 2021-10-12 19:34:24
'''
import os
import re
import sys
import json
import yaml
import threading
from threading import current_thread
from kafka import KafkaConsumer, TopicPartition, OffsetAndMetadata

class ConsumerManager(threading.Thread):
    def __init__(self,
                 bootstrap_servers=['10.43.75.74:9092'],
                 topic='test_topic',
                 consumer_group_id='ci_conclude01'):
        '''
            初始化消费者
            topic: kafka topic，从kafka topic中订阅消息
        '''
        threading.Thread.__init__(self)
        self.topic = topic
        self.consumer = KafkaConsumer(group_id=consumer_group_id,
                                      max_poll_records=1,
                                      max_poll_interval_ms=60000,
                                      auto_offset_reset='earliest',
                                      bootstrap_servers=bootstrap_servers)
        self.topicPartition = TopicPartition(topic=topic, partition=1)
        self.consumer.assign([self.topicPartition])
        # self.consumer.seek_to_beginning()

        position = self.consumer.position(self.topicPartition)
        print(f"{consumer_group_id} offset: {position}")
        offset = self.consumer.beginning_offsets(self.consumer.assignment())
        print(f"{consumer_group_id} offset: {offset}")
        # from code import interact
        # interact(local=locals())

        self.consumer.seek(self.topicPartition, position)
    
    def parseMessage(self, message):
        value = json.loads(message)
        del value["data"]["file"]
        print(value)
        pass
    
    def run(self):
        '''
            执行订阅
            获取message
        '''
        thread_name = current_thread().getName()  # 获取线程名称
        print("thread name:%s" % thread_name)
        try:
            for message in self.consumer:
                print("#"*100)
                print("%s:%d:%d: key=%s value=%s" %
                            (message.topic, message.partition, message.offset,
                            message.key, message.value[:10]))
                try:
                    content = self.parseMessage(message.value)
                except Exception as e:
                    print(e)
        except Exception as e:
            print(e)
            # self.restart()
